package com.atguigu.flink.chapter10;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.*;

public class TableApiDemo1 {
    public static void main(String[] args) throws Exception {
        //表的执行
        /*EnvironmentSettings conf = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tEnv = TableEnvironment.create(conf);
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("s1", 1000L, 10),
                new WaterSensor("s2", 2000L, 20),
                new WaterSensor("s1", 3000L, 30),
                new WaterSensor("s1", 4000L, 40),
                new WaterSensor("s2", 5000L, 50)
        );

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        Table table = tEnv.fromDataStream(stream);
        table.printSchema();

        //api的方式：select * from t where id = 's1'
        Table result = table.where($("id").isEqual("s1"))
                .select($("id"), $("ts"), $("vc"));

        //把表转成流
        DataStream<WaterSensor> resultStream = tEnv.toDataStream(result, WaterSensor.class);
        resultStream.print();

        env.execute();
    }
}



//public class TableApiDemo1 {
//    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//
//        DataStreamSource<WaterSensor> stream = env.fromElements(
//                new WaterSensor("s1", 1000L, 10),
//                new WaterSensor("s2", 2000L, 20),
//                new WaterSensor("s3", 3000L, 30),
//                new WaterSensor("s4", 4000L, 40),
//                new WaterSensor("s5", 5000L, 50)
//        );
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//
//        Table table = tEnv.fromDataStream(stream);
//        table.printSchema();
//
//        //api的方式：select * from t where id = 's1'
//        Table result = table.where($("id").isEqual("s1"))
//                .select($("id"), $("ts"), $("vc"));
//
//        //把表转成流
//        DataStream<WaterSensor> resultStream = tEnv.toDataStream(result, WaterSensor.class);
//        resultStream.print();
//
//
//        env.execute();
//    }
//}



//public class TableApiDemo1 {
//    public static void main(String[] args) throws Exception {
//        //表的执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//        DataStreamSource<WaterSensor> stream = env.fromElements(
//                new WaterSensor("s1", 1000L, 10),
//                new WaterSensor("s2", 2000L, 20),
//                new WaterSensor("s3", 3000L, 30),
//                new WaterSensor("s4", 4000L, 40),
//                new WaterSensor("s5", 5000L, 50)
//        );
//
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//
//        Table table = tEnv.fromDataStream(stream);
//        table.printSchema();
//
//        //api的方式：select * from t where id = 's1'
//        Table result = table.where($("id").isEqual("s1"))
//                .select($("id"), $("ts"), $("vc"));
//
//
//        //把表转换成流
//        DataStream<WaterSensor> resultStream = tEnv.toDataStream(result, WaterSensor.class);
//        resultStream.print();
//
//        env.execute();
//    }
//}


//public class TableApiDemo1 {
//    public static void main(String[] args) throws Exception {
//        //表的执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//
//        DataStreamSource<WaterSensor> stream = env.fromElements(
//                new WaterSensor("s1", 1000L, 10),
//                new WaterSensor("s2", 2000L, 20),
//                new WaterSensor("s1", 3000L, 30),
//                new WaterSensor("s1", 4000L, 40),
//                new WaterSensor("s2", 5000L, 50)
//        );
//
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//
//        Table table = tEnv.fromDataStream(stream);
//        table.printSchema();
//
//        //api的方式：select * from t where id = 's1'
//        Table result = table.where($("id").isEqual("s1"))
//                .select($("id"), $("ts"), $("vc"));
//
//        //把表转成流
//        DataStream<WaterSensor> resultStream = tEnv.toDataStream(result, WaterSensor.class);
//        resultStream.print();
//
//        env.execute();
//    }
//}

//public class TableApiDemo1 {
//    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//
//        DataStreamSource<WaterSensor> stream = env.fromElements(
//                new WaterSensor("s1", 1000L, 10),
//                new WaterSensor("s2", 2000L, 20),
//                new WaterSensor("s1", 3000L, 30),
//                new WaterSensor("s1", 4000L, 40),
//                new WaterSensor("s2", 5000L, 50)
//        );
//
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//
//        Table table = tEnv.fromDataStream(stream);
//        table.printSchema();
//
//        //api的方式：select * from t where id = 's1'
//        Table result = table.where($("id").isEqual("s1"))
//                .select($("id"), $("ts"), $("vc"));
//
//        //把表转换成流
//        DataStream<WaterSensor> resultStream = tEnv.toDataStream(result, WaterSensor.class);
//        resultStream.print();
//
//        env.execute();
//    }
//}